package com.zsy.rocketconsumerdemo.consumer;

import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

import javax.annotation.PostConstruct;

@Setter
@Getter
public abstract class BaseConsumer {
    protected String tag = "";

    DefaultMQPushConsumer consumer;

    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer(getTag());
        consumer.setInstanceName(this.getTag());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setNamesrvAddr("47.100.22.19:9876");
        consumer.subscribe("zsyTopicTest", this.getTag());
        consumer.registerMessageListener(obtainMessageListener());
        consumer.start();
        System.out.println(this.getTag() + "Consumer启动。");
    }

    protected MessageListenerConcurrently obtainMessageListener() {
        return (msgs, context) -> {
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
            msgs.forEach(msg -> {
                System.out.println(String.format("Tag:%s", msg.getTags()));
                if (StringUtils.equalsAnyIgnoreCase(msg.getTags(), "zsyDelay")) {
                    System.out.println(String.format("delayTime:%d", System.currentTimeMillis() - msg.getBornTimestamp()));
                }
                System.out.println(String.format("Message:%s", new String(msg.getBody())));
            });
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        };
    }
}
